/*
 * [y] hybris Platform
 *
 * Copyright (c) 2000-2017 SAP SE
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of SAP
 * Hybris ("Confidential Information"). You shall not disclose such
 * Confidential Information and shall use it only in accordance with the
 * terms of the license agreement you entered into with SAP Hybris.
 */
package acc.asset.sync.job;

import de.hybris.deltadetection.ChangeDetectionService;
import de.hybris.deltadetection.ChangesCollector;
import de.hybris.deltadetection.ItemChangeDTO;
import de.hybris.deltadetection.StreamConfiguration;
import de.hybris.deltadetection.enums.ChangeType;
import de.hybris.deltadetection.enums.ItemVersionMarkerStatus;
import de.hybris.deltadetection.model.StreamConfigurationModel;
import de.hybris.platform.catalog.model.CatalogVersionModel;
import de.hybris.platform.core.model.type.ComposedTypeModel;
import de.hybris.platform.cronjob.enums.CronJobResult;
import de.hybris.platform.cronjob.enums.CronJobStatus;
import de.hybris.platform.servicelayer.cronjob.AbstractJobPerformable;
import de.hybris.platform.servicelayer.cronjob.PerformResult;
import de.hybris.platform.servicelayer.cronjob.TypeAwareJobPerformable;
import de.hybris.platform.servicelayer.media.MediaService;
import de.hybris.platform.servicelayer.search.FlexibleSearchQuery;
import de.hybris.platform.servicelayer.search.SearchResult;
import de.hybris.platform.tx.Transaction;
import de.hybris.platform.tx.TransactionBody;
import de.hybris.platform.util.Config;
import de.hybris.platform.util.logging.Logs;
import de.hybris.y2ysync.deltadetection.collector.ItemTypeGroupingCollectorWithBatching;
import de.hybris.y2ysync.model.Y2YColumnDefinitionModel;
import de.hybris.y2ysync.model.Y2YStreamConfigurationContainerModel;
import de.hybris.y2ysync.model.Y2YStreamConfigurationModel;
import de.hybris.y2ysync.model.Y2YSyncCronJobModel;
import de.hybris.y2ysync.model.Y2YSyncJobModel;
import de.hybris.y2ysync.task.internal.MediasForType;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;

import com.google.common.collect.Lists;

import acc.asset.sync.DemoSyncTaskFactory;


/**
 *
 */
public class DemoY2YSyncMasterJobPerformable extends AbstractJobPerformable<Y2YSyncCronJobModel>
		implements TypeAwareJobPerformable
{

	private static final Logger LOG = Logger.getLogger(DemoY2YSyncMasterJobPerformable.class);
	private static final int DEFAULT_BATCH_SIZE = 100;
	private static final String COLUMNS_DELIMITER = ";";
	private ChangeDetectionService changeDetectionService;
	private MediaService mediaService;
	private DemoSyncTaskFactory syncTaskFactory;

	@Override
	public String getType()
	{
		return null;
	}

	@Override
	public boolean createDefaultJob()
	{
		return false;
	}

	@Override
	public PerformResult perform(final Y2YSyncCronJobModel cronJob)
	{
		final Y2YSyncJobModel job = cronJob.getJob();
		final Y2YStreamConfigurationContainerModel streamConfigurationContainer = job.getStreamConfigurationContainer();
		if (streamConfigurationContainer == null)
		{
			LOG.info("CronJob: " + cronJob.getCode() + " configuration container is missing");
			return new PerformResult(CronJobResult.ERROR, CronJobStatus.ABORTED);
		}

		final ArrayList<MediasForType> allMedias = Lists.newArrayList();
		final HashMap<String, Object> globalQueryParameters = new HashMap<String, Object>();
		this.fillParameters(globalQueryParameters, streamConfigurationContainer.getCatalogVersion());
		final Set<StreamConfigurationModel> configurations = this.getActiveConfigurations(streamConfigurationContainer);
		if (configurations.isEmpty())
		{
			LOG.info("CronJob: " + cronJob.getCode() + " configuration container is empty");
			return new PerformResult(CronJobResult.ERROR, CronJobStatus.ABORTED);
		}

		for (final StreamConfigurationModel streamConfiguration : configurations)
		{
			final Y2YStreamConfigurationModel y2yStreamConfiguration = (Y2YStreamConfigurationModel) streamConfiguration;
			final StreamConfiguration configuration = this.toStreamConfiguration(globalQueryParameters, y2yStreamConfiguration);
			LOG.info("configuration parameters:" + configuration.getParameters());
			LOG.info("configuration itemSelector:" + configuration.getItemSelector());


			final FlexibleSearchQuery fq = createFlexibleSearchQuery(streamConfiguration.getItemTypeForStream(), configuration);
			LOG.info("Query to be executed: \n" + fq + "\n");

			final ItemTypeGroupingCollectorWithBatching collector = new ItemTypeGroupingCollectorWithBatching(cronJob.getCode(),
					this.getBatchSize(), this.modelService, this.mediaService);

			collecItems(fq, collector, configuration);

			final MediasForType.MediasForTypeBuilder mediasForTypeBuilder = MediasForType.builder()
					.withComposedTypeCode(y2yStreamConfiguration.getItemTypeForStream().getCode())
					.withImpExHeader(
							this.join(y2yStreamConfiguration, y2YColumnDefinitionModel -> y2YColumnDefinitionModel.getImpexHeader()))
					.withDataHubColumns(
							this.join(y2yStreamConfiguration, y2YColumnDefinitionModel -> y2YColumnDefinitionModel.getColumnName()))
					.withMediaPks(collector.getCreatedMedias())
					.withDataHubType(StringUtils.defaultString(y2yStreamConfiguration.getDataHubType()));
			final MediasForType mediasForType = mediasForTypeBuilder.build();

			LOG.info("MediasForType composedTypeCode###impexHeader:" + mediasForType.getComposedTypeCode() + "###" + "###"
					+ mediasForType.getImpexHeader() + "\n\n");
			allMedias.add(mediasForType);
		}

		try
		{
			Transaction.current().execute(new TransactionBody()
			{

				@Override
				public Void execute() throws Exception
				{
					getSyncTaskFactory().runSyncTasks(cronJob.getCode(), job.getSyncType(), allMedias, job.getNodeGroup());
					return null;
				}
			});
		}
		catch (final Exception e)
		{
			throw new IllegalStateException("Exception occurred while creating tasks", e);
		}

		return new PerformResult(CronJobResult.SUCCESS, CronJobStatus.RUNNING);

	}

	private void fillParameters(final Map<String, Object> globalQueryParameters, final CatalogVersionModel catalogVersion)
	{
		if (catalogVersion == null)
		{
			return;
		}
		globalQueryParameters.put("catalog", catalogVersion.getCatalog());
		globalQueryParameters.put("catalogVersion", catalogVersion);
		globalQueryParameters.put("catalogVersionCode", catalogVersion.getVersion());
		final String catalogName = catalogVersion.getCatalog().getName();
		if (StringUtils.isNotEmpty(catalogName))
		{
			globalQueryParameters.put("catalogName", catalogName);
		}
	}

	private Set<StreamConfigurationModel> getActiveConfigurations(final Y2YStreamConfigurationContainerModel container)
	{
		return container.getConfigurations().stream()
				.filter(streamConfigurationModel -> streamConfigurationModel.getActive().booleanValue()).collect(Collectors.toSet());
	}

	private StreamConfiguration toStreamConfiguration(final Map<String, Object> globalQueryParameters,
			final Y2YStreamConfigurationModel y2YStreamConfigurationModel)
	{
		final HashMap<String, Object> streamQueryParameters = new HashMap<String, Object>(globalQueryParameters);
		this.fillParameters(streamQueryParameters, y2YStreamConfigurationModel.getCatalogVersion());
		return StreamConfiguration.buildFor(y2YStreamConfigurationModel.getStreamId())
				.withItemSelector(y2YStreamConfigurationModel.getWhereClause())
				.withVersionValue(y2YStreamConfigurationModel.getVersionSelectClause())
				.withExcludedTypeCodes(y2YStreamConfigurationModel.getExcludedTypes()).withParameters(streamQueryParameters);
	}

	private int getBatchSize()
	{
		return (int) Config.getLong("y2ysync.batch.size", 5000);
	}

	private String join(final Y2YStreamConfigurationModel y2yStreamConfiguration,
			final Function<Y2YColumnDefinitionModel, String> mapper)
	{
		final String columns = y2yStreamConfiguration.getColumnDefinitions().stream()
				.sorted((def1, def2) -> def1.getPosition().compareTo(def2.getPosition())).map(mapper)
				.collect(Collectors.joining(";"));
		Logs.debug(LOG, () -> "StreamConfig Columns string for type: " + y2yStreamConfiguration.getItemTypeForStream().getCode()
				+ " >>> " + columns);
		return columns;
	}

	public FlexibleSearchQuery createFlexibleSearchQuery(final ComposedTypeModel composedType,
			final StreamConfiguration configuration)
	{
		final String query = this.prepareQueryForFindingChangesByType(composedType, configuration);
		final FlexibleSearchQuery fq = new FlexibleSearchQuery(query);
		fq.addQueryParameter("streamId", configuration.getStreamId());
		fq.addQueryParameter("statusActive", ItemVersionMarkerStatus.ACTIVE);
		fq.addQueryParameters(configuration.getParameters());
		final Integer useVersionValue = StringUtils.isBlank(configuration.getVersionValue()) ? Integer.valueOf(0)
				: Integer.valueOf(1);
		fq.addQueryParameter("useVersionValue", useVersionValue);
		fq.setResultClassList(
				Arrays.asList(Long.class, ComposedTypeModel.class, Date.class, String.class, String.class, String.class));
		return fq;
	}

	private String prepareQueryForFindingChangesByType(final ComposedTypeModel composedType,
			final StreamConfiguration configuration)
	{
		final StringBuilder typePksAsQueryParams = new StringBuilder();
		typePksAsQueryParams.append(composedType.getPk().getLong());

		final Set excludedTypeCodes = configuration.getExcludedTypeCodes();
		composedType.getAllSubTypes().stream().filter(type -> !excludedTypeCodes.contains(type.getCode())).forEach(type -> {
			typePksAsQueryParams.append(",").append(type.getPk().getLong());
		});

		final String typeCode = composedType.getCode();
		final String itemSelector = configuration.getItemSelector();
		final String versionValue = configuration.getVersionValue();
		//		final String queryForDeleted = this.getQueryForDeletedItems(typePksAsQueryParams.toString(), typeCode, itemSelector);
		//		final String queryForModified = this.getQueryForModifiedItems(typePksAsQueryParams.toString(), typeCode, itemSelector,
		//				versionValue);
		final String queryForNew = this.getQueryForAllItems(typePksAsQueryParams.toString(), typeCode, itemSelector, versionValue);
		final String combinedQuery = "SELECT t.itemPK, t.itemComposedType, t.version, t.changeType, t.info, t.versionValue FROM( {{"
				+ queryForNew + "}} )t";
		return combinedQuery;
	}


	private String getQueryForDeletedItems(final String typePksAsQueryParams, final String baseType, final String itemSelector)
	{
		final String wherePart = StringUtils.isBlank(itemSelector) ? "" : " AND (" + itemSelector + ")";
		return String.format(
				"SELECT {ivm.itemPK} AS itemPK, {ivm.itemComposedType} AS itemComposedType, {ivm.versionTS} AS version, 'DELETED' AS changeType, {ivm.info} AS info, {ivm.versionValue} AS versionValue FROM {ItemVersionMarker AS ivm} WHERE {ivm.itemComposedType} IN(%s) AND {ivm.streamId}=?streamId AND {ivm.status}=?statusActive AND NOT EXISTS({{SELECT {item.PK} FROM {%s as item} WHERE {ivm.itemPK}={item.PK}%s}})",
				typePksAsQueryParams, baseType, wherePart);
	}

	private String getQueryForModifiedItems(final String typePksAsQueryParams, final String baseType, final String itemSelector,
			final String versionValuePart)
	{
		final String wherePart = StringUtils.isBlank(itemSelector) ? "" : " AND (" + itemSelector + ")";
		final String versionValue = StringUtils.isBlank(versionValuePart) ? "('1')" : "(" + versionValuePart + ")";
		return String.format(
				"SELECT {item.PK} AS itemPK, {item.itemType} AS itemComposedType, {item.modifiedTime} AS version, 'MODIFIED' AS changeType, {ivm.info} AS info, %s AS versionValue FROM {%s AS item JOIN ItemVersionMarker AS ivm ON {ivm.itemPK}={item.PK}} WHERE {item.itemType} IN(%s)%s AND {ivm.streamId}=?streamId AND {ivm.status}=?statusActive AND ((?useVersionValue=1 AND ({ivm.versionValue}<>%s OR ({ivm.versionValue} is null AND %s is not null) OR ({ivm.versionValue} is not null AND %s is null))) OR (?useVersionValue<>1 AND {item.modifiedTime}>{ivm.versionTS}))",
				versionValue, baseType, typePksAsQueryParams, wherePart, versionValue, versionValue, versionValue);
	}

	private String getQueryForNewItems(final String typePksAsQueryParams, final String baseType, final String itemSelector,
			final String versionValuePart)
	{
		final String wherePart = StringUtils.isBlank(itemSelector) ? "" : " AND (" + itemSelector + ")";
		final String versionValue = StringUtils.isBlank(versionValuePart) ? "('1')" : "(" + versionValuePart + ")";
		return String.format(
				"SELECT {item.PK} AS itemPK, {item.itemType} AS itemComposedType, {item.modifiedTime} AS version, 'NEW' AS changeType, 'created' AS info, %s AS versionValue FROM {%s AS item} WHERE {item.itemType} IN(%s)%s AND NOT EXISTS({{SELECT {ivm.PK} FROM {ItemVersionMarker as ivm} WHERE {ivm.itemPK}={item.PK} AND {ivm.streamId}=?streamId AND {ivm.status}=?statusActive}})",
				versionValue, baseType, typePksAsQueryParams, wherePart);
	}

	private String getQueryForAllItems(final String typePksAsQueryParams, final String baseType, final String itemSelector,
			final String versionValuePart)
	{
		final String wherePart = StringUtils.isBlank(itemSelector) ? "" : " AND (" + itemSelector + ")";
		final String versionValue = StringUtils.isBlank(versionValuePart) ? "('1')" : "(" + versionValuePart + ")";
		return String.format(
				"SELECT {item.PK} AS itemPK, {item.itemType} AS itemComposedType, {item.modifiedTime} AS version, 'NEW' AS changeType, 'created' AS info, %s AS versionValue FROM {%s AS item} WHERE {item.itemType} IN(%s)%s ",
				versionValue, baseType, typePksAsQueryParams, wherePart);

		//		+ "AND NOT EXISTS("
		//		+ "	{{"
		//		+ "	SELECT {ivm.PK} FROM {ItemVersionMarker as ivm} WHERE {ivm.itemPK}={item.PK} AND {ivm.streamId}=?streamId AND {ivm.status}=?statusActive"
		//		+ "	}}"
		//		+ ")"
	}

	private void collecItems(final FlexibleSearchQuery fq, final ChangesCollector collector,
			final StreamConfiguration configuration)
	{
		long count = 0;
		final SearchResult<List> result = this.flexibleSearchService.search(fq);
		try
		{
			for (final List row : result.getResult())
			{
				if (row == null || collector.collect(getChangeFromRow(row, configuration.getStreamId())))
				{
					count++;
					continue;
				}
				LOG.info("Collecting changes stopped");
				break;
			}
		}
		finally
		{
			collector.finish();
		}

		LOG.info("record collected:" + count);

	}

	private ItemChangeDTO getChangeFromRow(final List<Object> row, final String streamId)
	{
		final String itemComposedType = ((ComposedTypeModel) row.get(1)).getCode();
		final Long itemPK = (Long) row.get(0);
		final Date version = (Date) row.get(2);
		final ChangeType changeType = ChangeType.valueOf(((String) row.get(3)).trim());
		final String info = (String) row.get(4);
		final String versionValue = (String) row.get(5);
		return new ItemChangeDTO(itemPK, version, changeType, info, itemComposedType, streamId).withVersionValue(versionValue);
	}

	/**
	 * @return the changeDetectionService
	 */
	public ChangeDetectionService getChangeDetectionService()
	{
		return changeDetectionService;
	}

	/**
	 * @param changeDetectionService
	 *           the changeDetectionService to set
	 */
	public void setChangeDetectionService(final ChangeDetectionService changeDetectionService)
	{
		this.changeDetectionService = changeDetectionService;
	}

	/**
	 * @return the mediaService
	 */
	public MediaService getMediaService()
	{
		return mediaService;
	}

	/**
	 * @param mediaService
	 *           the mediaService to set
	 */
	public void setMediaService(final MediaService mediaService)
	{
		this.mediaService = mediaService;
	}

	/**
	 * @return the syncTaskFactory
	 */
	public DemoSyncTaskFactory getSyncTaskFactory()
	{
		return syncTaskFactory;
	}

	/**
	 * @param syncTaskFactory
	 *           the syncTaskFactory to set
	 */
	public void setSyncTaskFactory(final DemoSyncTaskFactory syncTaskFactory)
	{
		this.syncTaskFactory = syncTaskFactory;
	}



}
